package com.yuda.tools.mq.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

@Controller
public class MSNAliYun {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    private static final int mQThreadNum = 10;

    private final ExecutorService executorService = Executors.newFixedThreadPool(mQThreadNum);

    private final CloudAccount account = new CloudAccount("LTAIssc9TCX1Pe0Q", "oCmWCl1vClDXhdyR9KWPe90CqJcl6P",
            "http://1098879781219972.mns.cn-beijing-internal.aliyuncs.com"); //.cn-beijing-internal.aliyuncs.com

    //this client need only initialize once
    private final MNSClient client = account.getMNSClient();

    private CloudQueue queue = null;

    AtomicLong atomicLong = new AtomicLong(0);

    private final AtomicReference<Short> powerSupply = new AtomicReference<>(null);


    @ResponseBody
    @RequestMapping("/msnStart")
    public void receiver() {
        try {
            //防止一下开了若干个任务。
            if (powerSupply.get() == null) {
                queue = client.getQueueRef("stuanswer");
                //先用一个线程去探测
                executorService.execute(() -> {
                    for (int i = 0; i < Integer.MAX_VALUE; i++) {
                        Message message = queue.peekMessage();
                        if (message != null && message.getMessageId() != null) {
                            //取原来的值看看是不是1 ，如果是1不需要设置
                            Short aShort = powerSupply.get();
                            if (null == aShort || aShort == 0) {
                                System.out.println("任务状态为： 开启");
                                powerSupply.set((short) 1);
                                //通知线程执行
                                executorService.execute(this::executorTask);
                            }
                        } else {
                            powerSupply.set((short) 0);
                            System.out.println("任务状态为： 关闭");
                        }
                        try {
                            TimeUnit.SECONDS.sleep(2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
            } else {
                System.out.println("有任务正在运行中...");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void executorTask() {
        CountDownLatch countDownLatch = new CountDownLatch(8);
        for (int j = 0; j < 8; j++) {
            executorService.execute(() -> {
                while (powerSupply.get() > 0) {
                    Message popMsg = queue.popMessage();
                    if (popMsg != null) {
                        System.out.println(atomicLong.getAndIncrement());
                        try {
                            JSONObject jo = JSON.parseObject(popMsg.getMessageBodyAsRawString());
                            System.out.println(jo.toJSONString());
                            System.out.println(jo.getInteger("time"));
                            int update = jdbcTemplate.update("insert into mq_answer (exam_uuid,student_id,subject_id,subject_relation_id,ques_id,paper_id,answer,answer_json,md5,`type`,ques_type,create_time) values (?,?,?,?,?,?,?,?,?,?,?,?)",
                                    jo.getString("examUuid"), jo.getString("studentId"), jo.getString("subjectId"), jo.getString("subjectRelationId"),
                                    jo.getString("questionId"), jo.getString("paperId"), jo.getString("answer"), jo.getString("answerJson"),
                                    jo.getString("md5"), jo.getInteger("type"), jo.getInteger("questype"), jo.getInteger("time"));
                            if (update == 1) {
                                queue.deleteMessage(popMsg.getReceiptHandle());
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            System.err.println("解析json 失败...");
                        }

                    }
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
            powerSupply.set((short)0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @ResponseBody
    @RequestMapping("/clearMsnMq")
    public void clearMsnMq() {
        //this client need only initialize once
        MNSClient client = account.getMNSClient();
        //循环消费10条消息
        try {
            CloudQueue queue = client.getQueueRef("stuanswer");
            for (int i = 0; i < 5000; i++) {
                Message popMsg = queue.popMessage();
                if (popMsg != null) {
                    queue.deleteMessage(popMsg.getReceiptHandle());
                }
            }
        } catch (Exception e) {
            System.out.println("Unknown exception happened!");
            e.printStackTrace();
        }
        client.close();
    }
}
